package com.mooc.kafka.admin;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.clients.admin.AlterConfigsResult;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.CreatePartitionsResult;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author StarsOfFuture_xYang
 * @version 1.0
 * @date 2021-04-03 5:24 下午
 * @information kafka-study - com.mooc.kafka.admin
 **/
public class AdminSimple {
    public static final String TOPIC_NAME="test_admin_2";

    public static void main(String[] args)throws Exception {
        createTopic();
//        listTopics();
//        delTopic();
//        listTopics();
//        describeConfigTopic();
//        alterTopicConfig();
//        describeTopic();
//        addPartitions(3);
//        describeTopic();
    }

    /**
     * 添加Partitions
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void addPartitions(int total) throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        HashMap<String, NewPartitions> stringNewPartitionsHashMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(total);
        stringNewPartitionsHashMap.put(TOPIC_NAME,newPartitions);
        CreatePartitionsResult partitions = adminClient.createPartitions(stringNewPartitionsHashMap);
        System.out.println(partitions.all().get());
        adminClient.close();
    }

    /**
     * 获取topic配置信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void describeTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        describeTopicsResult.all().get().entrySet().stream().forEach(entry->{
            System.out.println("k: "+entry.getKey()+" v:"+entry.getValue());
        });
    }


    /**
     * 修改配置信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void alterTopicConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        HashMap<ConfigResource, Collection<AlterConfigOp>> map = new HashMap<>();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        AlterConfigOp alterConfigOp = new AlterConfigOp(new ConfigEntry("preallocate", "true"), AlterConfigOp.OpType.SET);
        map.put(configResource,Arrays.asList(alterConfigOp));
        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(map);
        System.out.println(alterConfigsResult.all().get());
        adminClient.close();
    }

    /**
     * 获取Topic配置信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void describeConfigTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        describeConfigsResult.all().get().entrySet().stream().forEach(entry->{
            System.out.println("key: "+entry.getKey()+"  value: "+entry.getValue());
        });

    }

    /**
     * 删除topic
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void delTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        DeleteTopicsResult test = adminClient.deleteTopics(Arrays.asList("test"));
        System.out.println(test.all().get());
        adminClient.close();
    }

    /**
     * 查询listTopics
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void listTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
        listTopicsResult.names().get().stream().forEach(System.out::println);
        adminClient.close();
    }

    /**
     * 创建Topic
     * @throws Exception
     */
    public static void createTopic()throws Exception{
        AdminClient adminClient = getAdminClient();
        short rs=1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME,1,rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println(topics.all().get());
        adminClient.close();
    }

    /**
     * 创建adminClient连接
     * @return
     */
    public static AdminClient getAdminClient(){
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"1.116.23.168:9092");
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }
}


class FxAdmin{

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        createTopic();
//        listTopics();
//        getTopicConfig();
//        describeConfig();
        createPartitions();
    }

    public static void createPartitions() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        HashMap<String, NewPartitions> stringNewPartitionsHashMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(2);
        stringNewPartitionsHashMap.put(TOPIC_NAME,newPartitions);
        CreatePartitionsResult partitions = adminClient.createPartitions(stringNewPartitionsHashMap);
        System.out.println(partitions.all().get());
    }

    public static void describeConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        System.out.println(describeConfigsResult.all().get());
        adminClient.close();
    }

    public static void getTopicConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        System.out.println(describeTopicsResult.all().get());
        adminClient.close();
    }


    public static void listTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
        listTopicsResult.names().get().stream().forEach(System.out::println);
        adminClient.close();
    }

    private static final String TOPIC_NAME="test_222";

    public static void createTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        short rs=1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME, 1, rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println(topics.all().get());
        adminClient.close();
    }

    public static AdminClient getAdminClient(){
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"1.116.23.168:9092");
        return AdminClient.create(properties);
    }
}
